package com.archive.core.executor;

import com.archive.core.processor.ArchiveProcessor;
import com.archive.core.table.ArchiveColumn;
import com.archive.core.table.ArchiveTable;
import com.archive.core.util.ArchiveUtil;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.List;
import java.util.Map;

public class SameDBInsertIntoSelectArchiveExecutor extends AbstractArchiveExecutor {

    public SameDBInsertIntoSelectArchiveExecutor(ArchiveProcessor archiveProcessor) {
        super(archiveProcessor);
    }

    @Override
    public void checkArchiveTable(String tableName, String archiveTableName, Connection connection,
                                  Connection targetConnection) throws SQLException {
        checkArchiveTableSame(tableName, archiveTableName, connection);
    }

    @Override
    public void appendQueryColumns(ArchiveTable archiveTable, StringBuilder sql, boolean appendShortName, String shortName) {
        List<String> extractColumns = archiveTable.getExtractColumns();
        sql.append("select ");
        for (String column : extractColumns) {
            if (appendShortName) {
                sql.append(shortName).append(".");
            }
            sql.append(column).append(",");
        }
        sql.deleteCharAt(sql.length() - 1);
    }

    @Override
    public Object extractQueryData(ArchiveTable archiveTable, Object maxPrimaryValue, ResultSet resultSet, List<String> extractColumns,
                                   Map<String, List<Object>> extractDataMap, List<Object[]> data) throws SQLException {
        while (resultSet.next()) {
            int columnCount = extractColumns.size();
            Object[] rowData = new Object[columnCount];
            for (int i = 0; i < columnCount; i++) {
                String columnName = extractColumns.get(i);
                Object columnValue = resultSet.getObject(i + 1);
                rowData[i] = columnValue;

                extractDataMap.get(columnName).add(columnValue);
            }
            data.add(rowData);
        }
        int size = data.size();
        if (size > 0) {
            maxPrimaryValue = data.get(size - 1)[archiveTable.getPrimaryKeyIndex()];
        }
        return maxPrimaryValue;
    }

    @Override
    protected int getNormalInsertSql(ArchiveTable archiveTable, String targetTableName, StringBuilder insertSql) {
        insertSql.append("insert into ").append(targetTableName).append(" select * from ")
                .append(archiveTable.getTableName()).append(" where ").append(archiveTable.getPrimaryKey()).append(" in(");
        return insertSql.length();
    }

    @Override
    protected void appendNormalInsertValue(StringBuilder insertSql, Object[] datum) {
        appendColumnValue(insertSql, datum[0]);
    }

    @Override
    protected void addInsertBatch(Statement insertStatement, StringBuilder insertSql, int insertSqlClearPoint) throws SQLException {
        addDeleteBatch(insertStatement, insertSql, insertSqlClearPoint);
    }

    @Override
    protected StringBuilder getPreparedInsertSql(ArchiveTable archiveTable, String archiveTableName) {
        return new StringBuilder("insert into ").append(archiveTableName).append(" select * from ")
                .append(archiveTable.getTableName()).append(" where ").append(archiveTable.getPrimaryKey()).append(" in(");
    }

    @Override
    protected Statement getPreparedInsertStatement(Connection connection, Connection targetConnection, StringBuilder insertSql) throws SQLException {
        return connection.createStatement();
    }

    @Override
    public void checkTableColumn(String tableName, String archiveTableName) throws SQLException {
        Connection connection = null;
        Statement statement = null;
        Connection targetConnection = null;
        Statement targetStatement = null;
        try {
            String checkTable = "'" + tableName + "'" + ",'" + archiveTableName + "'";
            // 表结构修改所需时长未知，使用原始数据库连接，数据源获取的连接可能导致超时
            connection = getConnection();

            statement = connection.createStatement();
            Map<String, Map<String, ArchiveColumn>> tableColumns = getColumns(connection.getCatalog(), checkTable, statement);
            checkAndModifyColumns(tableName, archiveTableName, tableColumns, statement);
        } finally {
            ArchiveUtil.closeConnection(connection);
            ArchiveUtil.closeStatement(statement);
            ArchiveUtil.closeConnection(targetConnection);
            ArchiveUtil.closeStatement(targetStatement);
        }
    }

    @Override
    protected void appendPreparedInsertValue(Statement insertStatement, Object[] row, StringBuilder insertSql, int batchSize, Integer executeBatchSize, int clearPoint) throws SQLException {
        appendColumnValue(insertSql, row[0]);
        if (batchSize % executeBatchSize == 0) {
            addDeleteBatch(insertStatement, insertSql, clearPoint);
        }
    }

    @Override
    protected void appendPreparedRestInsertValue(Statement insertStatement, StringBuilder insertSql, int clearPoint) throws SQLException {
        addDeleteBatch(insertStatement, insertSql, clearPoint);
    }

}
